// Copyright 2023 Srhino Co., Ltd. All rights reserved.

package v1

import (
	"context"
	"errors"
	"git.ouryun.cn/lzj1/grpc-bidirectional-stream/pkg/apis/v1"
	v12 "git.ouryun.cn/lzj1/grpc-bidirectional-stream/pkg/interfaces/v1"
	"git.ouryun.cn/lzj1/grpc-bidirectional-stream/pkg/utils"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/protobuf/proto"
	"io"
	"log"
	"net"
	"sync/atomic"
	"time"
)

var reqId int64

type grpcServer struct {
	v1.UnimplementedStreamDataServiceServer
	address    string
	streamData *StreamData
	stream     v1.StreamDataService_UserStreamServer
	callback   RequestProcessFunc
	ctx        context.Context
}

type RequestProcessFunc func(message *v1.StreamMessage)

var _ v12.GrpcServer = (*grpcServer)(nil)

func NewGrpcServer(addr string, callback RequestProcessFunc) v12.GrpcServer {
	return &grpcServer{
		address:    addr,
		streamData: NewStreamData(),
		callback:   callback,
	}
}

func (s *grpcServer) UserStream(stream v1.StreamDataService_UserStreamServer) error {
	defer func() {
		if r := recover(); r != nil {
			log.Println(r)
		}
	}()
	log.Println("client connection is successful")

	// 获取客户端的唯一标识符，可以根据实际需求进行处理
	value := stream.Context().Value("client-id")
	log.Printf("client id is %v", value)
	log.Printf("client id is %v", stream.Context())
	//md, ok := metadata.FromIncomingContext(stream.Context())
	//marshal, _ := json.Marshal(md)
	//log.Printf("client id is %s", marshal)
	//if !ok {
	//	return status.Errorf(codes.InvalidArgument, "missing metadata")
	//}
	//
	//id := md.Get("client-id")
	//if len(id) == 0 {
	//	return status.Errorf(codes.InvalidArgument, "missing client-id in metadata")
	//}
	//
	//log.Printf("client id is %s", id[0])

	s.stream = stream

	// 接收客户端数据
	err := s.handleRecv()
	if err != nil {
		return err
	}
	return nil
}

// 接收grpc客户端数据
func (s *grpcServer) handleRecv() error {
	for {
		recv, err := s.stream.Recv()
		log.Printf("接收到客户端数据 %v", recv)
		// 错误处理
		if err != nil {
			log.Println(err)
			if grpc.Code(err) == codes.Canceled {
				s.stream = nil
			}

			if err == io.EOF {
				s.stream = nil
			}
			return err
		}
		// 客户端数据处理
		if recv.Type == v1.StreamMessage_RESPONSE_TYPE {
			err := s.handleResponse(recv)
			if err != nil {
				return err
			}
		} else {
			s.handleRequest(recv)
		}
	}
}

// 处理grpc客户端响应数据
func (s *grpcServer) handleResponse(recv *v1.StreamMessage) error {
	log.Printf("handleResponse %v", recv)
	s.streamData.Responses <- recv
	return nil
}

// 处理grpc客户端请求数据
func (s *grpcServer) handleRequest(recv *v1.StreamMessage) {
	log.Printf("handleRequest %v", recv)
	if s.callback != nil {
		s.callback(recv)
	}
}

// 处理转发请求
func (s *grpcServer) forwardRequest() {
	for {
		select {
		case request := <-s.streamData.Requests:
			log.Println(request)
			if err := s.sendMessage(request); err != nil {
				log.Println(err)
			}
		}
	}

}

// 发送消息到grpc客户端
func (s *grpcServer) sendMessage(message *v1.StreamMessage) error {
	if s.stream == nil {
		return errors.New("stream closed")
	}
	log.Printf("发送消息 %v", message)
	err := s.stream.Send(message)

	if err != nil {
		return err
	}

	return nil
}

// Request 接收转发请求
func (s *grpcServer) Request(data proto.Message, params *v1.QueryParams, messageType v1.StreamMessage_Type, requestId uint64) (*v1.StreamMessage, error) {
	dataAny, err := utils.MessageToAny(data)
	if err != nil {
		return nil, err
	}

	message := &v1.StreamMessage{
		Type:   messageType,
		Data:   dataAny,
		Params: params,
	}

	if messageType == v1.StreamMessage_REQUEST_TYPE {
		atomic.AddInt64(&reqId, 1)
		message.RequestId = uint64(reqId)
	} else {
		message.RequestId = requestId
	}

	s.streamData.Requests <- message
	if messageType == v1.StreamMessage_REQUEST_TYPE {
		for {
			select {
			case <-time.After(5 * time.Second):
				return nil, errors.New("request timeout")
			case msg := <-s.streamData.Responses:
				// TODO 这里的处理方式还需要优化
				if msg.RequestId == message.RequestId {
					return msg, nil
				}
			}
		}
	}
	return nil, nil
}

func (s *grpcServer) Start() error {
	go func() {
		s.forwardRequest()
	}()

	listener, _ := net.Listen("tcp", s.address)
	grpcserver := grpc.NewServer()
	// 注册服务方法
	v1.RegisterStreamDataServiceServer(grpcserver, s)

	if err := grpcserver.Serve(listener); err != nil {
		log.Println(err)
		return err
	}

	return nil
}

func (s *grpcServer) Stop() {
	// TODO: 关闭连接
}
